/*
 * for i in {1..1000}; do echo "---- $i ----"; mpiexec -np 2 ./mppbf; done
 * 
 * File:   ub6_tbb_mpi.cpp
 * Author: sergey
 *
 * Created on 18 Декабрь 2010 г., 14:54
 *
 * Это программа создаётся для использования на сервере Amazon Web Services.
 * Она использует технологии: OpenMPI, Threading Building Blocks и Boost Serialize.
 *
 * Цель этой программы: найти все 6-местные функции сложности 13 (количество
 * слагаемых в полиномиальных представлениях термов этих функций).
 * Для того, чтобы найти все эти функции приходится перебирать пары 5-местных
 * функций.
 *
 * Перебираются следующие пары функций:
 * (7,7), (7,8), (7,9), (8,8), (8,9) и (9,9).
 * Где левое число $\alpha$ пары означает, что надо перебрать все 5-местные
 * функции сложности $\alpha$. А правое число $\beta$ означает, что надо
 * перебрать все 5-местные функции сложности $\beta$.
 *
 * Чтобы оценить размерность перебора этих пар, приводятся следующие данные:
 *   L       LP              Все
 *   7    2 143    1 299 295 404
 *   8       86       11 460 744
 *   9        2            7 824
 *
 * Таким образом, суммарный объем перебора будет 2 809 953 504 492 функций.
 * При нашем бюджете необходимо сделать так, чтобы производительность составляла
 * 545 940 064 функуций/час (или 151 650 функций/секунду или 
 * 18 956 функций/unit*секунду). Такая оценка делается при стоимости $0,68 за
 * один час работы Extra-Large High-CPU On-Demand Instance на AWS.
 *
 * В данный момент можно предположить, что такие расчеты вполне справедливы
 * и вполне можно ожидать получение результатов в указанных рамках (правда,
 * в надежде, что многие функции не будут нуждаться в минимизации).
 *
 * ---
 * Итак, вычисления происходят на каждом узле. Больше всего работы достается
 * в итоге главному узлу, так как он помимо прочего управляет всем приложением
 * и генерирует (раздрает) новые задачи, а так же сохраняет состояние всего
 * приложения на случай аварийного завершения.
 *
 * 2 784 390 050 772 функций в паре 7х7. 
 * 150 270 в секунду на одном компьютере
 */

#include "DefineConfiguration.h"
#include <cstdlib>
#include <iostream>
#include <tbb/blocked_range.h>
#include <tbb/parallel_for.h>
#include <tbb/compat/thread>
#include <tbb/tick_count.h>
#include "tbb/spin_mutex.h"
#include "tbb/mutex.h"
#include <mpi/mpi.h>
#include <string>
#include <sstream>
//#include <boost/archive/text_oarchive.hpp>
//#include <boost/archive/text_iarchive.hpp>
#include <fstream>
#include <time.h>
#include <vector>
#include <queue>

#include <csignal>

#include "LibraryInterface.h"
#include "BooleanFunction.h"
#include "NLLibrary.h"
#include "LPLibrary.h"
#include "G6Library.h" 

//#define DEBUG_MPI

bool continueProcessing = true; // Продолжаем выполнение?

struct MPITags {
    typedef int type;
    enum {
        NoCommand,
        FinishProcessing,
        NewTask,
        ResultSize,
        ResultArray
    };
};

class SystemPars {
public:
//    friend class boost::serialization::access;
    int MPI_size; 
    UInt32 tgf;

    SystemPars() {}
//    template<class Archive>
//    void serialize(Archive & ar, const unsigned int version) {
//        ar & MPI_size;
//        ar & tgf;
//    }
} parObject = SystemPars();

/*
 * Это класс класс управления приложением для главного узла. Данный класс
 * является некоторым командным интерпретатором.
 */

bool yesno(std::string text) {
    std::cout << "Are you sure " << text << "? (y/n): ";
    char c;
    std::cin >> c;
    if (c == 'y') return true;
    else return false;
}

class ConcurentTasksContainer {
private:
    std::vector<UInt64> minTasks; // Массив, в котором хранится отправленная задача минимизации
    size_t minTasksNum; // количество занятых минимизацией узлов

    std::vector<MPITags::type> commands; // массив комманд
    size_t commandsNum; // количество команд

    tbb::spin_mutex qMutex;

    std::vector<bool> haveSendedFP; // был ли отправлен FinishProcessing
    size_t nFinishedNodes; // количество не завершившихся узлов

    bool exitState;

public:
    ConcurentTasksContainer() {
        minTasksNum = 0;
        qMutex = tbb::spin_mutex();
        nFinishedNodes = 0;
        exitState = false;
    }
    void resize(size_t size) {
        minTasks.resize(size);
        commands.resize(size);
        haveSendedFP.resize(size);
        nFinishedNodes = size;
    }

    bool hasCommands() {
        tbb::spin_mutex::scoped_lock l(qMutex);
        return commandsNum > 0;
    }

    bool hasCommand(size_t node) {
        tbb::spin_mutex::scoped_lock l(qMutex);
        return commands[node] != MPITags::NoCommand;
    }

    void setCommand(size_t node, MPITags::type com) {
        tbb::spin_mutex::scoped_lock l(qMutex);
        if (!haveSendedFP[node] && commands[node] == MPITags::NoCommand) {
            commands[node] = com;
            commandsNum++;
        }
        if (com == MPITags::FinishProcessing) {
            haveSendedFP[node] = true;
        }
    }

    MPITags::type getCommand(size_t node) {
        tbb::spin_mutex::scoped_lock l(qMutex);
        MPITags::type com = commands[node];
        if (commands[node] == MPITags::FinishProcessing) {
            nFinishedNodes--;
        }
        if (commands[node] != MPITags::NoCommand) {
            commands[node] = MPITags::NoCommand;
            commandsNum--;
        }
        return com;
    }

    void newTask(size_t node, UInt64 task) {
        tbb::spin_mutex::scoped_lock l(qMutex);
        if (!haveSendedFP[node]) {
            minTasks[node] = task;
            minTasksNum++;
        }
    }

    void endTask(size_t node) {
        tbb::spin_mutex::scoped_lock l(qMutex);
        minTasks[node] = 0;
        minTasksNum--;
    }

    bool isEmpty() {
        tbb::spin_mutex::scoped_lock l(qMutex);
        return minTasksNum == 0 && commandsNum == 0;
    }

    bool isProcessing() {
        tbb::spin_mutex::scoped_lock l(qMutex);
        return nFinishedNodes != 0;
    }

    void exit() {
        this->exitState = true;
        for(int i = 1; i < parObject.MPI_size; i++)
            setCommand(i, MPITags::FinishProcessing);
    }

    bool isExit() {
        return exitState;
    }

} taskContainer;

class ConsoleHandlerThread {
public:
    void operator ()() {
        printf("ConsoleHandlerThread: start\n");
        std::string s;
        while(true) {
            std::cin >> s;
            if (s == "exit") {
                if (yesno("want to exit")) {
                    if (taskContainer.hasCommands()) {
                        printf("Some of nodes still have unprocessed commands. Wait for a while.");
                    }
                    taskContainer.exit();
                }
            } else {
                printf("Unknown command.\n");
            }
        }
        printf("ConsoleHandlerThread: end\n");
    }

    ConsoleHandlerThread() { }
};

class TaskGenerator {
private:
    bool hsr, hn;
    NLLibrary<LibraryInfo<5> > *lib;
    inline void findNext() {
        UInt8 c;
        do {
            if (f == 0xffffffff) {
                f = 0;
                hn = false;
                return ;
            }
            f++;
            lib->get(f,c);
        } while (c != 7);
        hn = true;
    }
    UInt32 tG;
public:
    UInt32 f;
    TaskGenerator(NLLibrary<LibraryInfo<5> > &nllib) {
        f = 0xfffee993UL;
        lib = &nllib;
        hsr = false;
        tG = 0; // tasks generated
    }
    UInt32 getTG() {
        return tG;
    }
    bool hasNext() {
        if (!hsr) {
            hsr = true;
            findNext();
            return hn;
        }
        return hn;
    }
    UInt32 nextTask() {
        tG++;
        if (hsr) {
            hsr = false;
            return f;
        }

        if (hasNext()) {
            hsr = false;
            return f;
        } else {
            tG--;
            hsr = false;
            return 0;
        }
    }
};


NLLibrary<LibraryInfo<5> > nllib;
/*
 * Поток генерации новых функций и отправки их всем узлам кластера.
 */
void serverProcessing() {
    printf("serverProcessing: begin\n");

    std::ostringstream filename;
    std::ofstream resultFile;

    filename << LIBRARYDIRECTORIES << "/nllibs/nllibrary_5.bin";
    nllib.load(filename.str());
    
    resultFile.open("result.bin", std::ios::out | std::ios::binary | std::ios::app);

    TaskGenerator taskGenerator(nllib);
    taskGenerator.f = parObject.tgf - 1;
    
    printf("start from func %#x\n", taskGenerator.f);

    srand(time(NULL));
    taskContainer.resize(parObject.MPI_size);
    MPI_Status state;
    UInt64 f;
    UInt32 FunctionsNum;
    std::vector<UInt64> resarray;
    UInt64 totalRAsize = 0;

    for(int i = 1; i < parObject.MPI_size && taskGenerator.hasNext(); i++) {
        f = taskGenerator.nextTask();
        MPI_Send(&f, 1, MPI_UNSIGNED_LONG_LONG, i, MPITags::NewTask, MPI_COMM_WORLD);
        taskContainer.newTask(i, f);
        #ifdef DEBUG_MPI
        printf("serverProcessing: present to %d\n",i);
        #endif
    }

    while ((!taskContainer.isExit() || taskContainer.hasCommands()) && taskGenerator.hasNext()) {
        MPI_Recv(&FunctionsNum, 1, MPI_UNSIGNED_LONG, MPI_ANY_SOURCE, MPITags::ResultSize, MPI_COMM_WORLD, &state);
        resarray.resize(FunctionsNum);
        totalRAsize += FunctionsNum;
        MPI_Recv(&resarray[0], FunctionsNum, MPI_UNSIGNED_LONG_LONG, state.MPI_SOURCE, MPITags::ResultArray, MPI_COMM_WORLD, &state);
        
        resultFile.write((char*)&resarray[0], resarray.size() * sizeof(UInt64));
        
        taskContainer.endTask(state.MPI_SOURCE);
        #ifdef DEBUG_MPI
        printf("serverProcessing: result from %d \n",state.MPI_SOURCE);
        #endif

        if (taskContainer.hasCommand(state.MPI_SOURCE)) {
            MPI_Send(&f, 1, MPI_UNSIGNED_LONG_LONG, state.MPI_SOURCE, taskContainer.getCommand(state.MPI_SOURCE), MPI_COMM_WORLD);
            #ifdef DEBUG_MPI
            printf("serverProcessing: sent message to %d\n", state.MPI_SOURCE);
            #endif
        } else {
            f = taskGenerator.nextTask();
            MPI_Send(&f, 1, MPI_UNSIGNED_LONG_LONG, state.MPI_SOURCE, MPITags::NewTask, MPI_COMM_WORLD);
            taskContainer.newTask(state.MPI_SOURCE, f);
            #ifdef DEBUG_MPI
            printf("serverProcessing: sent to %d\n", state.MPI_SOURCE);
            #endif
        }
    }

    // пока все не пришлют - получаем результаты
    while(!taskContainer.isEmpty()) {
        MPI_Recv(&FunctionsNum, 1, MPI_UNSIGNED_LONG, MPI_ANY_SOURCE, MPITags::ResultSize, MPI_COMM_WORLD, &state);
        resarray.resize(FunctionsNum);
        totalRAsize += FunctionsNum;
        MPI_Recv(&resarray[0], FunctionsNum, MPI_UNSIGNED_LONG_LONG, state.MPI_SOURCE, MPITags::ResultArray, MPI_COMM_WORLD, &state);
        
        resultFile.write((char*)&resarray[0], resarray.size() * sizeof(UInt64));

        #ifdef DEBUG_MPI
        printf("serverProcessing: postreceived from %d \n",state.MPI_SOURCE);
        #endif
        taskContainer.endTask(state.MPI_SOURCE);

        MPI_Send(&f, 1, MPI_UNSIGNED_LONG_LONG, state.MPI_SOURCE, MPITags::FinishProcessing, MPI_COMM_WORLD);
    }
    resultFile.close();
    parObject.tgf = taskGenerator.f;
    printf("serverProcessing: end\n  stop at f1 %#x\n  total f1 %d\n  total result size %d (%dKiB)\n", 
        taskGenerator.f,
        taskGenerator.getTG(),
        totalRAsize,
        totalRAsize / 64
    );
}

std::vector<UInt32> lplibSel;
G6Library<NLLibrary<LibraryInfo<5> >, GParamsExample, GParamsAlgsExample> *g6;

tbb::spin_mutex MinByLPSelMutex;

// Счетчи "быстрой" минимизации.
UInt32 Sin = 0, Sout = 0;

class MinByLPSel {
    UInt32 f1g;
    std::vector<UInt64> *result;
public:
    void operator() (const tbb::blocked_range<size_t>& r) const {
        UInt64 f, fs;
        UInt32 fp, f1, f2;
        UInt8 c, c1, c2;
        for(size_t i = r.begin(); i != r.end(); i++) {
            Sin++;

            f = (f1g << 32) ^ lplibSel[i];

            /* первая "быстрая" оценка */
            nllib.get(f1g ^ lplibSel[i], c);
            if (c < 6 || c > 7) continue;

            /* вторая "быстрая" оценка */
            BooleanFunction<LibraryInfo<6> >::transposition<0,UInt64>(f);
            f1 = f >> 32;
            f2 = f & 0xffffFFFF;
            fp = f1 ^ f2;
            nllib.get(f1, c1);
            nllib.get(f2, c2);
            nllib.get(fp, c);
            if (c1 + c2 < 13 || c1 + c < 13 || c2 + c < 13 || c1 > 7 || c2 > 7 || c > 7) continue;

            /* третья "быстрая" оценка */
            BooleanFunction<LibraryInfo<6> >::transposition<1,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<0,UInt64>(f);
            f1 = f >> 32;
            f2 = f & 0xffffFFFF;
            fp = f1 ^ f2;
            nllib.get(f1, c1);
            nllib.get(f2, c2);
            nllib.get(fp, c);
            if (c1 + c2 < 13 || c1 + c < 13 || c2 + c < 13 || c1 > 7 || c2 > 7 || c > 7) continue;

            /* четвертая "быстрая" оценка */
            BooleanFunction<LibraryInfo<6> >::transposition<2,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<1,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<0,UInt64>(f);
            f1 = f >> 32;
            f2 = f & 0xffffFFFF;
            fp = f1 ^ f2;
            nllib.get(f1, c1);
            nllib.get(f2, c2);
            nllib.get(fp, c);
            if (c1 + c2 < 13 || c1 + c < 13 || c2 + c < 13 || c1 > 7 || c2 > 7 || c > 7) continue;

            /* пятая "быстрая" оценка */
            BooleanFunction<LibraryInfo<6> >::transposition<3,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<2,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<1,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<0,UInt64>(f);
            f1 = f >> 32;
            f2 = f & 0xffffFFFF;
            fp = f1 ^ f2;
            nllib.get(f1, c1);
            nllib.get(f2, c2);
            nllib.get(fp, c);
            if (c1 + c2 < 13 || c1 + c < 13 || c2 + c < 13 || c1 > 7 || c2 > 7 || c > 7) continue;

            /* шестая "быстрая" оценка */
            BooleanFunction<LibraryInfo<6> >::transposition<4,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<3,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<2,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<1,UInt64>(f);
            BooleanFunction<LibraryInfo<6> >::transposition<0,UInt64>(f);
            f1 = f >> 32;
            f2 = f & 0xffffFFFF;
            fp = f1 ^ f2;
            nllib.get(f1, c1);
            nllib.get(f2, c2);
            nllib.get(fp, c);
            if (c1 + c2 < 13 || c1 + c < 13 || c2 + c < 13 || c1 > 7 || c2 > 7 || c > 7) continue;

            Sout++;
            g6->get(f, c, i);
            if (c >= 13) {
                tbb::spin_mutex::scoped_lock l(MinByLPSelMutex);
                result->push_back(f);
            }
        }
    }    
    MinByLPSel(const UInt32 f1, std::vector<UInt64> &result) : f1g(f1), result(&result) { }
};
/*
 * Эта функция запускается в основном потоке приложения и она получает от
 * главного узла сообщения с новыми задачами, которые выполняет и отправляет
 * результат обратно.
 */
void tasksProcessing(const int rank) {
    std::ostringstream filename;
    filename << LIBRARYDIRECTORIES << "/nllibs/nllibrary_5.bin";
    nllib.load(filename.str());

    filename.str("");

    filename << LIBRARYDIRECTORIES << "/lplibs/lplibrary_5.bin";
    std::fstream lplib(filename.str().c_str(), std::ios::in | std::ios::binary);
    while(!lplib.eof()) {
        UInt64 f;
        UInt8 c;
        lplib.read((char*)&f, sizeof(UInt64));
        lplib.read((char*)&c, sizeof(UInt8));
        if (c == 7) {
            lplibSel.push_back(f);
        }
    }
    lplib.close();
    filename.str("");
    
    g6 = new G6Library<NLLibrary<LibraryInfo<5> >, GParamsExample, GParamsAlgsExample>(nllib, lplibSel.size());

    MPI_Status state;
    UInt32 resultSize;
    std::vector<UInt64> result;
    UInt64 f1;

    srand(time(NULL));
    printf("tasksProcessing %d: begin\n", rank);
    result.reserve(lplibSel.size());

    bool continueProcessing = true;

    UInt32 i;

    while(continueProcessing) {
        #ifdef DEBUG_MPI
        printf("tasksProcessing %d: waiting\n", rank);
        #endif
        
        MPI_Recv(&f1, 1, MPI_UNSIGNED_LONG_LONG, 0, MPI_ANY_TAG, MPI_COMM_WORLD, &state);
        
        #ifdef DEBUG_MPI
        printf("tasksProcessing %d: have a message\n", rank);
        #endif

        switch (state.MPI_TAG) {
            case MPITags::NewTask:
                result.clear();
                tbb::parallel_for(tbb::blocked_range<size_t>(0, lplibSel.size(), 16), MinByLPSel(f1, result));
                resultSize = result.size();
                
                #ifdef DEBUG_MPI
                printf("tasksProcessing %d: minimization: sending\n", rank);
                #endif
                
                MPI_Send(&resultSize, 1, MPI_UNSIGNED_LONG, 0, MPITags::ResultSize, MPI_COMM_WORLD);
                MPI_Send(&result[0], resultSize, MPI_UNSIGNED_LONG_LONG, 0, MPITags::ResultArray, MPI_COMM_WORLD);
                break;
            case MPITags::FinishProcessing:
                continueProcessing = false;
                break;
        }

    }
    
    printf("tasksProcessing %d end:\n  minimized total %d\n  ""fast"" minimized %d(%d%)\n  total f1 %d\n", 
        rank, 
        Sin, 
        Sin - Sout, 100*(Sin - Sout) / Sin,
        Sin / lplibSel.size()
    );
}

void loadSettings(const int size) {
    std::ifstream ifs("settings");
    if (!ifs.fail()) {
        ifs >> parObject.MPI_size >> parObject.tgf;
    }
    parObject.MPI_size = size;
}

void saveSettings() {
    std::ofstream ofs("settings");
    ofs << parObject.MPI_size << ' ' << parObject.tgf;
}

std::thread *consoleHandlerThread;

void term_handler(int i) {
    std::cout << "Terminating..." << std::endl;
    consoleHandlerThread->detach();
    taskContainer.exit();
}

int main(int argc, char** argv) {
    MPI_Init(&argc, &argv);
    int rank, size;
    MPI_Comm_rank(MPI_COMM_WORLD,&rank);
    MPI_Comm_size(MPI_COMM_WORLD,&size);

    if (rank == 0) {
        signal(SIGTERM, term_handler);
        signal(SIGABRT, term_handler);
        loadSettings(size);
        
        consoleHandlerThread = new std::thread(ConsoleHandlerThread());

        serverProcessing();

        saveSettings();

        if (consoleHandlerThread->joinable())
            consoleHandlerThread->join();
    } else {
        tasksProcessing(rank);
    }
    
    MPI_Finalize();
    return 0;
}
